package com.heima.simple;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class ProducerQuertStart {

    public static void main(String[] args) {
        //2.连接kafka以及配置
        Properties properites=new Properties();
        //连接kafka
        properites.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.128:9092");
        //设置key和value字符串序列化
        properites.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properites.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        //设置重试的机制
        properites.put(ProducerConfig.RETRIES_CONFIG,3);
        //1.创建生产者对象
        KafkaProducer<String,String> kafkaProducer=new KafkaProducer<String, String>(properites);


        for (int i = 0; i <5 ; i++) {
            if(i%2==0){
                //3.执行发送消息
                ProducerRecord<String, String> record=new ProducerRecord<>("topic_kafka_start","key-first","hello kafka");
                //同步发送
                kafkaProducer.send(record);
            }else{
                //3.执行发送消息
                ProducerRecord<String, String> record=new ProducerRecord<>("topic_kafka_start","key-first","hello itcast");
                //同步发送
                kafkaProducer.send(record);
            }
        }





       /* Future<RecordMetadata> future = kafkaProducer.send(record);
        try {
            RecordMetadata recordMetadata = future.get();
            System.out.println("消息的偏移量:"+recordMetadata.offset());
        } catch (Exception e) {
            e.printStackTrace();
        }*/

       /* //异步发送
        kafkaProducer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                //比如重试
                if(exception!=null){
                    System.out.println(exception.getMessage());
                }
                //否则，发送成功，逻辑业务处理
                System.out.println("消息的偏移量:"+metadata.offset());

            }
        });*/


        //4.关闭资源
        kafkaProducer.close();
        System.out.println("生产者发送消息完成");
    }
}
